package com.copm.ifm.message.listenr;

import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.ObjectUtils;
import com.copm.ifm.base.basic.ServiceResponse;
import com.copm.ifm.base.basic.enums.ResponseCodeEnum;
import com.copm.ifm.common.redis.RedisUtil;
import com.copm.ifm.message.TemplateSend;
import com.copm.ifm.message.constance.RabbitMqCommon;
import com.copm.ifm.message.entity.SendTemplateDto;
import com.copm.ifm.message.entity.StaffReceive;
import com.copm.ifm.servers.sys.entity.NewTiming;
import com.copm.ifm.servers.sys.service.NewTimingService;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

/**
 * 消息监听消费消息任务表
 *
 * @author tp
 */
@Component
@Slf4j
public class RabbitMqListener {
    /**
     * Auto：
     * 1. 如果消息成功被消费（成功的意思是在消费的过程中没有抛出异常），则自动确认
     * 2. 当抛出 AmqpRejectAndDontRequeueException 异常的时候，则消息会被拒绝，且 requeue = false（不重新入队列）
     * 3. 当抛出 ImmediateAcknowledgeAmqpException 异常，则消费者会被确认
     * 4. 其他的异常，则消息会被拒绝，且 requeue = true，此时会发生死循环，可以通过 setDefaultRequeueRejected（默认是true）去设置抛弃消息
     * <p>
     * 如设置成manual手动确认，一定要对消息做出应答，否则rabbit认为当前队列没有消费完成，将不再继续向该队列发送消息。
     * <p>
     * channel.basicAck(long,boolean); 确认收到消息，消息将被队列移除，false只确认当前consumer一个消息收到，true确认所有consumer获得的消息。
     * <p>
     * channel.basicNack(long,boolean,boolean); 确认否定消息，第一个boolean表示一个consumer还是所有，第二个boolean表示requeue是否重新回到队列，true重新入队。
     * <p>
     * channel.basicReject(long,boolean); 拒绝消息，requeue=false 表示不再重新入队，如果配置了死信队列则进入死信队列。
     * <p>
     * 当消息回滚到消息队列时，这条消息不会回到队列尾部，而是仍是在队列头部，这时消费者会又接收到这条消息，如果想消息进入队尾，须确认消息后再次发送消息。
     * <p>
     */

    @Autowired
    NewTimingService newTimingService;

    @RabbitListener(queues = RabbitMqCommon.DELAYED_QUEUE_NAME)
    public ServiceResponse<?> receive(Message message, Channel channel) throws IOException {
        List<StaffReceive> receiveList = new ArrayList<>();
        //1.校验消息是否被消费
        if (message.getMessageProperties().getRedelivered()) {
            // 拒绝消息，requeue=false 表示不再重新入队，如果配置了死信队列则进入死信队列
            channel.basicReject(message.getMessageProperties().getDeliveryTag(), false);
            return new ServiceResponse<>(ResponseCodeEnum.SUCCESS);
        }
        String correlationId = message.getMessageProperties().getCorrelationId();
        //获取定时任务id（即将消费的任务）
        int timingId = Integer.parseInt(correlationId.replace(RabbitMqCommon.DELAYED_CORRELATION_ID_PRE, ""));
        NewTiming newsTiming = newTimingService.getOne(new LambdaQueryWrapper<NewTiming>().eq(NewTiming::getId, timingId));
        if (ObjectUtils.isEmpty(newsTiming)) {
            //这个定时任务如果为空则表示任务被删除或则已发送
            channel.basicReject(message.getMessageProperties().getDeliveryTag(), false);
        }

        //2.获取接收人信息
        String receiveUser = newTimingService.getOne(new LambdaQueryWrapper<NewTiming>()
                .eq(NewTiming::getTempId, timingId)).getReceiveUser();
        String[] split = receiveUser.split(",");
        List<Map<Object, Object>> mapList = new ArrayList<>();
        Arrays.asList(split).forEach(s -> {
            Map<Object, Object> hashEntries = RedisUtil.getInstance().getHashEntries(String.valueOf(newsTiming.getId()));
            mapList.add(hashEntries);
        });
        mapList.forEach(s -> {
            StaffReceive staffReceive = JSON.parseObject(JSON.toJSONString(s), StaffReceive.class);
            receiveList.add(staffReceive);
        });

        //3.封装消息体进行发送
        SendTemplateDto<StaffReceive> dto = new SendTemplateDto<>();
        dto.setTemplateId(newsTiming.getTempId());
        dto.setContent(newsTiming.getContent());
        dto.setSendType(1);
        //这里是已经包装成具体的接收人信息了
        dto.setReceiveUser(receiveList);
        TemplateSend templateSend = new TemplateSend();
        List<Integer> userList = receiveList.stream().map(StaffReceive::getId).collect(Collectors.toList());

        //4.消息发送
        try {
            ServiceResponse<?> send = templateSend.send(dto);
            //删除接收人
            userList.forEach(s -> RedisUtil.getInstance().delHash(String.valueOf(s)));
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), true);
            return send;
        } catch (IOException e) {
            channel.basicReject(message.getMessageProperties().getDeliveryTag(), false);
        }
        return new ServiceResponse<>(ResponseCodeEnum.SUCCESS);
    }

}
